package org.rcy.framework.mq.queue;

import org.rcy.framework.api.mq.AbstractMessage;
import org.rcy.framework.mq.timer.SystemTimer;
import org.rcy.framework.mq.timer.TimerTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author 吴康桥
 * @Description
 * @date 2022/11/23 17:44
 */


public class KafkaDelayQueue<M extends AbstractMessage> {
	private final Logger logger = LoggerFactory.getLogger(this.getClass());
	//驱动时间轮向前的线程
	private static ExecutorService executorService = Executors.newFixedThreadPool(1);
	private SystemTimer systemTimer;
	private KafkaTemplate template;

	public KafkaDelayQueue() {
	}

	public KafkaDelayQueue(SystemTimer systemTimer,KafkaTemplate template) {
		this.systemTimer = systemTimer;
		this.template = template;
	}

	public SystemTimer getSystemTimer() {
		return systemTimer;
	}

	public void setSystemTimer(SystemTimer systemTimer) {
		this.systemTimer = systemTimer;
	}

	public void add(String topic, M message,long delayTime) throws Exception {

		TimerTask timerTask = new TimerTask(delayTime) {
			@Override
			public void run() {
				logger.info("延迟消息即将发送");
				template.send(topic,message);
			}
		};
		systemTimer.add(timerTask);
		executorService.submit(() -> {
			while(true) {
				try {
					systemTimer.advanceClock(100L);
				} catch (Exception e) {
					logger.error(e.getMessage());
				}
			}
		});
	}
}
